\section{Distributing BLAST Jobs}


\subsection{The BLAST Application}

\subsubsection{Description}



BLAST is a collection of algorithms which identify similarities
between a genomic sequence and a database of other sequences.  When
BLAST is run, each query is compared against the specified database
according to the chosen algorithm, and the algorithm returns a the
names of sequences found in the database which are similar to those in
the input file.



Because BLAST is only an application of BioCompute, the original BLAST
algorithm itself is not modified.  The BLAST executable is used in its
original form, without any modifications or recompilations.  The focus
of BioCompute is to implement a generalized pattern of abstraction
atop Condor allowing for multiple applications to be expressed and
used by those with little or no background in formal computing.



\subsubsection{Inputs}
There are two input vectors for BLAST execution, hereafter referred to
as input files and databases.  BLAST input files are in the text-based
FASTA format specified by the NCBI.  They contain a list of queries,
and are supplied by the application user.  The beginning of each query
is noted with a \textbf{$\rangle$}, followed by the a description of
the sequence. The description typically includes a name and identifier
number, ended with a newline.  Sequence names and identification
numbers often follow conventions set by the NCBI.  A nucleotide or
peptide sequence follows, representing base pairs, amino acids, or
patterns thereof in a standardized text representation.  Whitespace
within the sequence is not significant, but lines shorter than 80
lines are recommended for readability.  A simple example of
FASTA-formatted file containing peptide sequences is shown in
Figure~\ref{samplepep}, while Figure~\ref{samplenuc} shows a
nucleotide sequence in FASTA format.  Typical input files for
BioCompute's initial target use cases range from several hundred bytes
to dozens of megabytes.

%todo:clean up section around names
%todo:NCBI reference: http://www.ncbi.nlm.nih.gov/blast/fasta.shtml
%todo: FASTA->formatdb->binary

BLAST databases are large compilations of sequences in a binary
format.  The databases used in BioCompute range in size from several
megabytes to several gigabytes.  Many databases are maintained and
hosted by the NBCI including the standard non-redundant database {\it
nr}.  Table~\ref{dblist} shows a list of 39 sequence databases currently
host by BioCompute, sourced from both the NCBI and datasets provided
by researchers at Notre Dame.


\begin{table}
\begin{center}
\begin{tabular}{|c|r|}
\hline
Name & Size \\
\hline
Heliconius\_e\_unigenes & 108K \\
ACWP2\_contigs\_v1 & 592K \\
ABWP1\_contigs\_v1 & 752K \\
CCMHS-1\_contigs\_v1 & 824K \\
CCWP1\_contigs\_v1 & 872K \\
ABWP2\_contigs\_v1 & 1.3M \\
ROB-2\_contigs\_v1 & 1.5M \\
WOA-2\_contigs\_v1 & 1.5M \\
ROA-2\_contigs\_v1 & 1.5M \\
WOB-1\_contigs\_v1 & 1.8M \\
Heliconius\_e\_proteins & 2.1M \\
ACHS1n\_contigs\_v1 & 2.2M \\
ACWP1\_contigs\_v1 & 2.3M \\
WOB-2\_contigs\_v1 & 2.4M \\
WOA-1\_contigs\_v1 & 3.0M \\
ACCanker\_contigs\_v1 & 3.2M \\
ROA-1\_contigs\_v1 & 3.4M \\
CCWP2\_contigs\_v1 & 3.7M \\
ROB-1\_contigs\_v1 & 3.9M \\
PfalciparumAllTranscripts\_PlasmoDB-5.5 & 4.1M \\
CCNHS-1\_contigs\_v1 & 5.1M \\
ACHS2n\_contigs\_v1 & 5.3M \\
CCCanker\_contigs\_v1 & 5.8M \\
1\_celera\_contigs & 6.3M \\
2\_celera\_contigs & 7.1M \\
bombyx\_predicted\_proteins & 7.2M \\
CCMHS-2\_contigs\_v1 & 11M \\
CCNHS-2\_contigs\_v1 & 13M \\
ButterflyBase & 18M \\
lepbase & 22M \\
Fagacea\_db.fasta & 30M \\
Fagaceae\_db & 87M \\
uniprot\_sprot & 233M \\
Drosophila\_melanogaster\_unigenes & 240M \\
agambiae.CHROMOSOMES-PEST.AgamP3 & 261M \\
pataa & 356M \\
patnt & 1.7G \\
uniprot\_trembl & 3.5G \\
uniprot\_trembl.fasta & 3.6G \\
nr & 6.4G \\
\hline
\end{tabular}

\caption{List of sequence databases hosted by BioCompute and sizes.}
\label{dblist}


\end{center}
\end{table}



\begin{figure}
\begin{verbatim}
>1397save_transcriptional_regulator_GCN4
VPESSDPAALKRARNTEAAR
RSRARKLQRMKQLEDKVEEL
LSKNYHLENEVARLKKLVGE
R
>1381save_cobratoxin
LECHNQQSSQTPTTTGCSGG
ETNCYKKRWRDHRGYRTERG
CGCPSVKNGIIINCCTTDRC
NN
\end{verbatim}
\caption{A sample peptide sequence in FASTA format}
\label{samplepep}
\end{figure}

\begin{figure}
\begin{verbatim}
>MAGI_43307_1
AAAGTTGAGTGGTTGGTAACA
GTTACCGTTCTTAAGATTGAT
CAACTATGGTGGTACGAGTCT
TGTAGAAAGTGCCTCAAGAAA
ACAAAGCCTCATGGTGATGCC
TATAAATGCTCGGACTCTGGT
TGTGGCCATGTTGGTCCGCCT
AATCCAAGGTACAGGTTGCTC
ATCACAGCAGGAGATGAGACAG
\end{verbatim}
\caption{A sample nucleotide sequence in FASTA format}
\label{samplenuc}
\end{figure}


\subsubsection{Outputs}


The original, unmodified BLAST program has many multiple output
formats which are not equivalent in form or content.  All of these
output options are presented to the BioCompute user.  The output of a
BioCompute BLAST job matches the output of an original BLAST in
content, but the form may be slightly different; depending on the
output type selected, additional program header information is
scattered throughout the file, one for each of the distributed Condor
jobs.  This output file could be normalized and scrubbed to more
closely represent the form of original BLAST queries, but this process would
be complicated by the myriad output options.  Figure~\ref{blastoutput}
shows an example of the default output option for BLAST.

\begin{figure}
\begin{verbatim}
Reference: Altschul, Stephen F., Thomas L. Madden, Alejandro A. Schaffer, 
Jinghui Zhang, Zheng Zhang, Webb Miller, and David J. Lipman (1997), 
"Gapped BLAST and PSI-BLAST: a new generation of protein database search
programs",  Nucleic Acids Res. 25:3389-3402.

Query= Contig2
         (910 letters)

Database: All non-redundant GenBank CDS
translations+PDB+SwissProt+PIR+PRF excluding environmental samples
from WGS projects 
           7,135,729 sequences; 2,462,332,163 total letters

Searching..................................................done

                                                                 Score    E
Sequences producing significant alignments:                      (bits) Value

ref|YP_002221402.1| NADH dehydrogenase subunit 2 [Blastocystis s...    37   2.1  
gb|ACD10933.1| NADH dehydrogenase subunit 2 [Blastocystis sp. Na...    37   2.1  
ref|YP_001096002.1| NADH dehydrogenase subunit 4 [Metaseiulus oc...    37   2.8  
ref|YP_001096013.1| NADH dehydrogenase subunit 4 [Metaseiulus oc...    37   2.8  
ref|ZP_00239675.1| hypothetical protein protein [Bacillus cereus...    36   4.7  

>ref|YP_002221402.1| NADH dehydrogenase subunit 2 [Blastocystis sp. NandII]
 gb|ACH86083.1| NADH dehydrogenase subunit 2 [Blastocystis sp. NandII]
          Length = 493

 Score = 37.0 bits (84), Expect = 2.1
 Identities = 24/81 (29%), Positives = 43/81 (53%), Gaps = 5/81 (6%)
 Frame = -1

Query: 829 FYHFKITFSLSDIDIVFDILLEIGENFIVYGLLYSALV----FLIMYYFVTYCVDNR-YI 665
           +++ KI + LSD+  V+     +G +FI+     + +     FL  +Y    C++N  Y+
Sbjct: 359 YFNNKIKY-LSDLSYVYKYNKTLGLSFIITMFSMAGVPPMAGFLAKFYSFFVCIENNFYL 417

Query: 664 LATLNDGLPKISNFYFVSFIK 602
           LAT+   L  I  FY++ F+K
Sbjct: 418 LATIGVLLSIICTFYYIRFLK 438
\end{verbatim}
\caption{A snippet of a typical BLAST output in the default format.}
\label{blastoutput}
\end{figure}


\subsection{Comparison to MapReduce}



A BLAST input file can have any number of query sequences.  Typical
jobs run during BioCompute's testing range from tens to hundreds of
thousands of queries.  Each query sequence is compared independently
against the specified database and the results of queries are not
dependent on the results of other queries.  Because of their
independence, these queries need not be compared against the database
in any particular order, nor need they be executed on the same
machine; the value of the function is determined by the input
sequence, the blast algorithm, and the database.


In order to distribute the queries among nodes, the original input
file is split into batches of N queries each.  Each of these resultant
input query shards now can be run against a BLAST database to produce
the sequences matching those in the input file.  This set of matching
sequences is a partial result for the query of the original input file
against the database, but it is a complete result for the query of the
input shard against the database.  These self-contained subsets of
results are then combined into one result, corresponding to the
result for the original input file.  See Figure~\ref{outline}.



\begin{figure}
\centering
\includegraphics[width=.5\columnwidth]{bc_outline.pdf}
\caption{Outline of job execution}
\label{outline}
\end{figure}

\begin{figure}
\centering
\includegraphics[width=.5\columnwidth]{bc_dataflow.pdf}
\caption{Data flow in BioCompute}
\label{dataflow}
\end{figure}

\begin{figure}
\centering
\includegraphics[width=.5\columnwidth]{bc_dividing_blast.pdf}
\caption{Divided input sequences run against a set of replicated databases}
\label{division}
\end{figure}


This model of dividing input and distributed execution closely
resembles Google's MapReduce model in some respects: a problem is
divided into independent subproblems, a map function is evaluated on
the input splits, and the intermediate values are reduced into a
result for the original problem.  In BioCompute, the input data is the
FASTA sequence file originating in the user, the map function is the
BLAST executable operating on an input shard and necessary databases,
and the reduce step is simply an ordered concatenation of the output
streams of the independent, distributed subproblems.

BioCompute differs from MapReduce in that it requires large, static
sequence databases, which are treated as atomic units to reduce
complexity in the problem domain.  This is discussed more in
Section~\ref{sec:atomicdb}.  MapReduce does not handle large,
indivisible files.  In order to provide these resources in whole on
remote execution nodes, BioCompute distributes these databases in
batch ahead of time, and therefore can assume their existence on
remote nodes when scheduling work processes.



\subsection{Execution Overview}

Each BLAST job submission to BioCompute is executed using the Notre
Dame Condor pool.  The following is the sequence of events which
occurs for every job submission.
\begin{enumerate}
\item The user uses the ``Submit Query'' page on the web interface to
select a local FASTA file to upload, choose a database to run the
sequence against, specify BLAST options, and name the job.  The user
submits the page and an HTTP post is submitted to the submission
processing page.
\item The submission processing page creates an entry in the database
for the job, creates a new job directory on the server, and copies the
sequence input file to said directory.  The current set of job
execution scripts is copied into the directory as well.
\item The job-execution script is called.  It's first task is to split
the sequence file into segments of N queries each. A Condor submit
file specifying the number of jobs to run and their respective inputs
is then generated.  A sample submit file is shown in
Figure~\ref{samplesubmit}.  At this point, the BioCompute job is ready
for submission to Condor: there are K input files of at most N queries
each, and the script which Condor will distribute is in the job
directory and ready to be copied to BioCompute worker nodes. 
\item The job is submitted to Condor.  Once jobs are dispatched to
  BioCompute computing nodes, the script checks for the existence of
  the required database on the local file-system.  If it is found, then
  the job will continue normally and BLAST will execute on the input
  split.  When a job is complete, its standard output and error
  streams are sent back to the master job directory and stored in the
  indicated filenames.
\item Upon completion of all jobs, the results from each job split are
  concatenated in order.  This is a linear time step, but if necessary
  in the future, this step can be easily divided as it is simply
  concatenation of ordered, numbered strings.  After this
  merge is completed, a timestamped ``completed'' file is written to the
  directory indicating the job is complete.  Optionally, an email can
  be sent to the user notifying them of the job's completion.
\item Upon a user requesting to view the job through BioCompute's web
  interface, the status of the job recorded in the database is
  checked.  If the job is not in a terminal state indicating its
  status will not change, i.e. ``Complete'' or ``Complete with
  Error'', the existence of the ``completed'' file in the master job
  directory is checked. If it is there, errors in execution are
  checked for, the database is duly updated and the newly-updated
  status is displayed to the user.
\end{enumerate}



%todo: check on email

\subsection{Submit File Overview}


The Condor submit file details execution details of a distributed job.
Many parts of a submit file are frequently similar across many
different types of jobs. This section will review the salient and
unique aspects of the sample submit file shown in
Figure~\ref{samplesubmit}.  This sample submit file is representative
of those generated for BioCompute jobs in general.


The first differentiating part of the submit file is the requirements
line.  This line is specifying what BioCompute requires of a node in
order to run.  The first clause specifies the machine group.
Currently, BioCompute nodes are restricted to the cluster designated
{\it sc0} within the Condor pool.  The next attribute restricts
BioCompute jobs to only run on the first CPU of a machine; under the
current configuration of the Condor pool, each CPU of a multi-cored
machine can be assigned jobs independently.  The final clause ensures
jobs are only assigned to machines where memory is currently
available.  These last two clauses were added because it was found
that BioCompute jobs would frequently deadlock when two jobs were
scheduled to two CPUs on the same machine.  Many of the sequence
databases in BioCompute are larger than available memory and as a
result the jobs are I/O-bound.  Scheduling two processes on the same
machine would result in detrimental competition for available memory
and disk access.  



The memory and CPU restrictions were added to reduce the possibility
of this happening.  This could result in reduced performance for jobs
against databases which are smaller than half the available memory
because otherwise jobs would be able to scheduled on both CPUs of a
machine without negative effects.  It should be emphasized here that
the {\it sc0} cluster is a shared cluster and not exclusive to processing
BioCompute jobs, and therefore there could be other processes running
and consuming memory on a machine.  The policy of restricting to a
single CPU has performed suitably so far for databases of all sizes. 


The next salient attribute of the submit file is the arguments line.
This is the set of command-line parameters which will be fed to
blast.script, a wrapper for the BLAST executable.  The parameters
include the command-line options which specify how the job will be
executed.  These include which variation of BLAST will be performed
(e.g. {\it blastn}, {\it blastp}, {\it tblastx}), which database to compare
against, the output format, and the number of sequences and alignments
to show. 

In addition to the BLAST job parameters, the BioCompute-assigned ID
number is passed along, and the Condor job number is indicated with the
\$(PROCESS) macro.  The Condor job number indicates which division of
the job is being executed and is in the range [0,Q), where Q is the
number on the {\it queue} line of the submit file.  The output and
error files which are transfered back to the master job directory are
marked with the Condor job ID to designate order and provide ensure unique
filenames.


The {\it on\_exit\_remove} line directs Condor how to handle a job
which fails to find the requisite database on the local node.  When
that happens, the script exits with error code 101, indicating to
Condor that it should be removed from that local node and reassigned
to another eligible node.


\subsection{Remote Execution Script Overview}
Three noteworthy events occur in the remote execution script on BioCompute nodes:

\begin{enumerate}
\item The script queries the local Chirp daemon for the local path of
files stored on the node.
\item The script checks for the existence and accessibility of the
database it will query against.
\item If the database is readable, the script calls the BLAST
executable.
\end{enumerate} 


The sequence databases are stored locally on the BioCompute nodes
through the Chirp distributed storage system.  Access to the Chirp file-system
is usually obtained through the {\it chirp\_server} process, but in
the special case where the requisite files stored via Chirp reside on
the local node, as is the case in BioCompute, faster access can be
obtained by accessing the local files directly and circumventing {\it
chirp\_server}.  To do this, the script requests the local Chirp path
from Chirp via the {\it localpath} directive and proceeds to access
the files through the local file-system.  In addition, to allow this,
the ACLs of the BioCompute database directory must be set to allow
local access.


The second item of note is that the script checks for the existence of
the database on the local node before calling the BLAST executable.
If it is not found, the script exits with code 101, directing Condor
to reschedule it to another node, as mentioned previously.

\begin{figure}
\begin{verbatim}
universe = vanilla
executable = ./blast.script
requirements = (MachineGroup=="sc0") && 
               (VirtualMachineID==1) && (Memory > 0)
arguments = 194 blastn 0 ButterflyBase 194.split.$(PROCESS) $(PROCESS) 50 
output = 194.output.$(PROCESS) 
error = 194.error.$(PROCESS) 
transfer_files = always 
transfer_input_file =194.split.$(PROCESS) 
on_exit_remove = (ExitCode!=101) || (ExitBySignal==TRUE) 
log = 194.logfile 
queue 2116

\end{verbatim}
\caption{A typical Condor submit file for BioCompute jobs}
\label{samplesubmit}
\end{figure}




\subsection{Reporting Job Status}
A Condor job generates and maintains a logfile in the master
submission directory.  Upon job completion, BioCompute generates a
file marking the time of completion.  Using these two sources of
information along with the size of the standard out stream, BioCompute
currently reports a job as Submitted, Running, Complete, or Complete
with Error.  A new system is being built which will rely solely on the
Condor logfile and give more granular status reporting.



\subsection{System Reliability}


BioCompute is entirely dependent on the underlying Condor system for
job execution.  Condor is a mature and proven distribution platform.
BioCompute should thus ``hand off'' jobs to Condor for execution as
quickly as possible in order to reduce the domain for errors and
potential for job failure.  By pushing responsibility to Condor
quickly, BioCompute reduces the possible space for job failure.  
